package cn.osworks.mqtt;

import java.io.IOException;
import java.sql.Timestamp;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

public class MqttTopicSubscribe implements MqttCallback{
	
	public static void main(String[] args) {
		int qos = 2;
		String broker = "tcp://localhost:1883";
		String subTopic = "Sample/Java";
		try {
			MqttTopicSubscribe mqttTopicSubscribe = new MqttTopicSubscribe(broker, "SampleJavaV3_Subscribe2");
			mqttTopicSubscribe.subscribe(subTopic, qos);
		} catch (MqttException me) {
			// Display full details of any exception that occurs
			System.out.println("reason " + me.getReasonCode());
			System.out.println("msg " + me.getMessage());
			System.out.println("loc " + me.getLocalizedMessage());
			System.out.println("cause " + me.getCause());
			System.out.println("excep " + me);
			me.printStackTrace();
		}
		
	}
	
	private MqttConnectOptions conOpt;
	private MqttClient client;
	public MqttTopicSubscribe(String brokerUrl, String clientId) throws MqttException {
		String tmpDir = System.getProperty("java.io.tmpdir");
		MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
		try {
			conOpt = new MqttConnectOptions();
			conOpt.setCleanSession(true);
			client = new MqttClient(brokerUrl, clientId, dataStore);
			client.setCallback(this);
		} catch (MqttException e) {
			e.printStackTrace();
			System.out.println("Unable to set up client: " + e.toString());
			System.exit(1);
		}
	}

	public void subscribe(String topicName, int qos) throws MqttException {

		client.connect(conOpt);
		System.out.println("Subscribing to topic \"" + topicName + "\" qos " + qos);
		client.subscribe(topicName, qos);
		System.out.println("Press <Enter> to exit");
		try {
			System.in.read();
		} catch (IOException e) {
			// If we can't read we'll just exit
		}
		client.disconnect();
		System.out.println("Disconnected");
	}

	@Override
	public void connectionLost(Throwable arg0) {
		
	}

	@Override
	public void deliveryComplete(IMqttDeliveryToken arg0) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void messageArrived(String topic, MqttMessage message) throws MqttException {
		// Called when a message arrives from the server that matches any
		// subscription made by the client
		String time = new Timestamp(System.currentTimeMillis()).toString();
		System.out.println("Time:\t" +time +
                           "  Topic:\t" + topic +
                           "  Message:\t" + new String(message.getPayload()) +
                           "  QoS:\t" + message.getQos());
	}

}
